package ex.tableapi.window;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.rowInterval;

public class ExBoundedOverWindows extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExBoundedOverWindows demo = new ExBoundedOverWindows();
        demo.getEnv();
        Table orders = demo.registerMysql2("myorder1", "orders1");
        Table result1= demo.exBoundedOverWindowsByproctime(orders);
        Table result2 = demo.exBoundedOverWindowsByrowtime(orders);

        demo.registerConsole(demo.createPrintOutDDL(result1.getResolvedSchema().toString()), demo.outName);
        result1.executeInsert(demo.outName);
    }

    private Table exBoundedOverWindowsByproctime(Table orders) {
        //proctime:按时a,c ,1分钟(rowtime:preceding(lit(1).minutes()))，分数据，(所有行都包含在窗口中)统计
        //proctime:按时a,c ,行数(rowInterval(3L))，分数据，累加统计
        orders.printSchema();
        Table result = orders.window(Over
                .partitionBy($("a"), $("c"))
                .orderBy($("pt"))
//                .preceding(lit(1).minutes())
                .preceding(rowInterval(3L))
                .as("w"))
                .select($("a"),
                        $("c"),
                        $("pt"),
                        $("b").count().over($("w")));
        return result;
    }

    private Table exBoundedOverWindowsByrowtime(Table orders) {
        //rowtime:按时a,c ,1分钟(rowtime:preceding(lit(1).minutes()))，分数据，累加统计
        //rowtime:按时a,c ,行数(rowInterval(3L))，分数据，累加统计
        orders.printSchema();
        Table result = orders.window(Over
                .partitionBy($("a"), $("c"))
                .orderBy($("rowtime"))
                .preceding(rowInterval(3L))
                .as("w"))
                .select($("a"),
                        $("c"),
                        $("rowtime"),
                        $("b").count().over($("w")));
        return result;
    }

}
